package avicit.bdp.dds.server.utils;

import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.dds.dispatch.enums.CommandType;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.enums.WarningType;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.dao.AlertDao;
import avicit.bdp.dds.dao.entity.Alert;
import avicit.bdp.dds.dao.entity.ProcessDefinition;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.platform6.api.sysmsg.dto.SysMsgDTO;
import avicit.platform6.api.system.SysMsgClient;
import avicit.platform6.core.context.ThreadContextHelper;
import avicit.platform6.core.restclient.RestProperties;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;

/**
 * @author admin
 * alert manager
 */
@Component
public class AlertManager {

	/**
	 * logger of AlertManager
	 */
	private static final Logger logger = LoggerFactory.getLogger(AlertManager.class);

	/**
	 * alert dao
	 */
	private final AlertDao alertDao = SpringApplicationContext.getBean(AlertDao.class);

	private SysMsgClient sysMsgClient = SpringApplicationContext.getBean(SysMsgClient.class);

	private RestProperties restProperties = SpringApplicationContext.getBean(RestProperties.class);


	/**
	 * command type convert chinese
	 *
	 * @param commandTypeCode command type code
	 * @return command name
	 */
	private String getCommandCnName(Integer commandTypeCode) {
		CommandType commandType = CommandType.of(commandTypeCode);
		switch (commandType) {
			case RECOVER_TOLERANCE_FAULT_PROCESS:
				return "recover tolerance fault process";
			case RECOVER_SUSPENDED_PROCESS:
				return "recover suspended process";
			case START_CURRENT_TASK_PROCESS:
				return "start current task process";
			case START_FAILURE_TASK_PROCESS:
				return "start failure task process";
			case START_PROCESS:
				return "start process";
			case REPEAT_RUNNING:
				return "repeat running";
			case SCHEDULER:
				return "scheduler";
			case COMPLEMENT_DATA:
				return "complement data";
			case PAUSE:
				return "pause";
			case STOP:
				return "stop";
			default:
				return "unknown type";
		}
	}

	/**
	 * process instance format
	 */
	private static final String PROCESS_INSTANCE_FORMAT =
			"\"id:%s\"," +
					"\"name:%s\"," +
					"\"job type: %s\"," +
					"\"state: %s\"," +
					"\"recovery:%s\"," +
					"\"run time: %d\"," +
					"\"start time: %s\"," +
					"\"end time: %s\"," +
					"\"host: %s\"";

	/**
	 * get process instance content
	 *
	 * @param processInstance process instance
	 * @param taskInstances   task instance list
	 * @return process instance format content
	 */
	private String getContentProcessInstance(ProcessInstance processInstance, List<TaskInstance> taskInstances) {
		String res = "";
		if (ExecutionStatus.of(processInstance.getState()).typeIsSuccess()) {
			res = String.format(PROCESS_INSTANCE_FORMAT,
					processInstance.getId(),
					processInstance.getName(),
					getCommandCnName(processInstance.getCommandType()),
					ExecutionStatus.of(processInstance.getState()).getDescp(),
					processInstance.getRecovery().toString(),
					processInstance.getRunTimes(),
					DateUtils.dateToString(processInstance.getStartTime()),
					DateUtils.dateToString(processInstance.getEndTime()),
					processInstance.getHost()

			);
			res = Constants.LEFT_SQUARE_BRACKET + res + Constants.RIGHT_SQUARE_BRACKET;
		} else if (ExecutionStatus.of(processInstance.getState()).typeIsFailure()) {

			List<LinkedHashMap> failedTaskList = new ArrayList<>();

			for (TaskInstance task : taskInstances) {
				if (ExecutionStatus.of(task.getState()).typeIsSuccess()) {
					continue;
				}
				LinkedHashMap<String, String> failedTaskMap = new LinkedHashMap<>();
				failedTaskMap.put("process instance id", String.valueOf(processInstance.getId()));
				failedTaskMap.put("process instance name", processInstance.getName());
				failedTaskMap.put("task id", String.valueOf(task.getId()));
				failedTaskMap.put("task name", task.getName());
				failedTaskMap.put("task type", task.getTaskType());
				failedTaskMap.put("task state", ExecutionStatus.of(task.getState()).getDescp());
				failedTaskMap.put("task start time", DateUtils.dateToString(task.getStartTime()));
				failedTaskMap.put("task end time", DateUtils.dateToString(task.getEndTime()));
				failedTaskMap.put("host", task.getHost());
				failedTaskMap.put("log path", task.getLogPath());
				failedTaskList.add(failedTaskMap);
			}
			res = JSONUtils.toJson(failedTaskList);
		}

		return res;
	}

	/**
	 * getting worker fault tolerant content
	 *
	 * @param processInstance   process instance
	 * @param toleranceTaskList tolerance task list
	 * @return worker tolerance content
	 */
	private String getWorkerToleranceContent(ProcessInstance processInstance, List<TaskInstance> toleranceTaskList) {

		List<LinkedHashMap<String, String>> toleranceTaskInstanceList = new ArrayList<>();

		for (TaskInstance taskInstance : toleranceTaskList) {
			LinkedHashMap<String, String> toleranceWorkerContentMap = new LinkedHashMap<>();
			toleranceWorkerContentMap.put("process name", processInstance.getName());
			toleranceWorkerContentMap.put("task name", taskInstance.getName());
			toleranceWorkerContentMap.put("host", taskInstance.getHost());
			toleranceWorkerContentMap.put("task retry times", String.valueOf(taskInstance.getRetryTimes()));
			toleranceTaskInstanceList.add(toleranceWorkerContentMap);
		}
		return JSONUtils.toJson(toleranceTaskInstanceList);
	}

	/**
	 * send worker alert fault tolerance
	 *
	 * @param processInstance   process instance
	 * @param toleranceTaskList tolerance task list
	 */
	public void sendAlertWorkerToleranceFault(ProcessInstance processInstance, List<TaskInstance> toleranceTaskList) {
		try {
			Alert alert = new Alert();
			alert.setTitle("worker fault tolerance");
			String content = getWorkerToleranceContent(processInstance, toleranceTaskList);
			alert.setContent(content);
			alert.setCreationDate(new Date());
			alert.setUserIds(processInstance.getWarningUserIds());
			alertDao.addAlert(alert);
			logger.info("add alert to db , alert : {}", alert.toString());

		} catch (Exception e) {
			logger.error("send alert failed:{} ", e.getMessage());
		}

	}

	/**
	 * send process instance alert
	 *
	 * @param processInstance process instance
	 * @param taskInstances   task instance list
	 */
	public void sendAlertProcessInstance(ProcessInstance processInstance,
										 List<TaskInstance> taskInstances) {
		logger.info("send alert info to users: " + processInstance.getWarningUserIds());
		boolean sendWarnning = false;
		WarningType warningType = WarningType.of(processInstance.getWarningType());
		switch (warningType) {
			case ALL:
				if (ExecutionStatus.of(processInstance.getState()).typeIsFinished()) {
					sendWarnning = true;
				}
				break;
			case SUCCESS:
				if (ExecutionStatus.of(processInstance.getState()).typeIsSuccess()) {
					sendWarnning = true;
				}
				break;
			case FAILURE:
				if (ExecutionStatus.of(processInstance.getState()).typeIsFailure()) {
					sendWarnning = true;
				}
				break;
			default:
		}
		if (!sendWarnning) {
			return;
		}

		Alert alert = new Alert();

		String cmdName = getCommandCnName(processInstance.getCommandType());
		String success = ExecutionStatus.of(processInstance.getState()).typeIsSuccess() ? "success" : "failed";
		alert.setTitle(cmdName + " " + success);
		//ShowType showType = ExecutionStatus.of(processInstance.getState()).typeIsSuccess() ? ShowType.TEXT : ShowType.TABLE;
		String content = getContentProcessInstance(processInstance, taskInstances);
		alert.setContent(content);
		alert.setUserIds(processInstance.getWarningUserIds());
		alert.setCreationDate(new Date());
		alert.setProcessDefinitionId(processInstance.getProcessDefinitionId());
		alert.setProcessInstanceId(processInstance.getId());
		alert.setCreatedBy(processInstance.getCreatedBy());
		alert.setAlertStatus(processInstance.getState());

		alertDao.addAlert(alert);
		logger.info("add alert to db , alert: {}", alert.toString());

        //消息发送到平台的消息管理
        String userIds = processInstance.getWarningUserIds();
        if (StringUtils.isBlank(userIds)) {
            return;
        }
        String[] ids = userIds.split(",");
        Date date = new Date();
        success = ExecutionStatus.of(processInstance.getState()).typeIsSuccess() ? "成功" : "失败";
        String sourceName = "离线计算通知";
        String sendUser = "1";
        ProcessDefinition definition = processInstance.getProcessDefinition();
        if (definition != null) {
            sourceName = getDefinitionTaskTypeName(definition);
            sendUser = definition.getCreatedBy();
        }
        for (String id : ids) {
            try {
                if (StringUtils.isBlank(id)) {
                    continue;
                }
                SysMsgDTO msg = new SysMsgDTO();
                msg.setTitle(processInstance.getName() + " 执行" + success);
                msg.setContent(content);
                msg.setSendUser(sendUser);
                msg.setRecvUser(id);
                msg.setSourceCode("personal");
                msg.setSourceName(sourceName);
                msg.setSendDate(date);
                msg.setMsgType("0");
                msg.setSendType("0");
                msg.setSysApplicationId(ThreadContextHelper.getAppId());
                sysMsgClient.sendmsg(msg);
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }

	}

	private String getDefinitionTaskTypeName(ProcessDefinition definition) {
		String name = "离线计算通知";
		if (definition.getType() == 0) {
			name = "数据同步通知";
		} else if (definition.getType() == 1) {
			name = "数据迁移通知";
		} else if (definition.getType() == 2) {
			switch (definition.getDatasourceId()) {
				case "commonFlow":
					name = "数据开发通知";
					break;
				case "sparkFlow":
					name = "离线计算通知";
					break;
				case "sparkStreamFlow":
					name = "实时计算通知";
					break;
				case "sparkMLFlow":
					name = "机器学习通知";
					break;
				default:
					name = "离线计算通知";
			}
		} else if (definition.getType() == 3) {
			name = "文件采集通知";
		} else {
			name = "消息通知";
		}

		return name;
	}

	/**
	 * send process timeout alert
	 *
	 * @param processInstance   process instance
	 * @param processDefinition process definition
	 */
	public void sendProcessTimeoutAlert(ProcessInstance processInstance, ProcessDefinition processDefinition) {
		alertDao.sendProcessTimeoutAlert(processInstance, processDefinition);
	}
}
